package cc.unmi;

import com.twitter.bijection.Injection;
import com.twitter.bijection.avro.GenericAvroCodecs;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @ref https://www.jianshu.com/p/a70950bab06d
 * @note 使用 Twitter 的 Bijection 类库实现 avro 的序列化与反序列化
 */
public class BijectionProducer {

    private KafkaProducer<String, byte[]> producer = new KafkaProducer<>(getProperties());

    private static String schemaStr = "{\n" +
            "    \"type\": \"record\",\n" +
            "    \"name\": \"User\",\n" +
            "    \"fields\": [\n" +
            "        {\"name\": \"id\", \"type\": \"int\"},\n" +
            "        {\"name\": \"name\",  \"type\": \"string\"},\n" +
            "        {\"name\": \"age\", \"type\": \"int\"}\n" +
            "    ]\n" +
            "}";

    public void sendData(Topic topic, int id, Integer age) {
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(schemaStr);
        Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(schema);

        GenericData.Record avroRecord = new GenericData.Record(schema);
        avroRecord.put("id", id);
        avroRecord.put("name", "name" + id);
        avroRecord.put("age", age);

        byte[] avroRecordBytes = recordInjection.apply(avroRecord);

        producer.send(new ProducerRecord<>(topic.topicName, avroRecordBytes),
                (metadata, exception) -> {
                    if (exception == null) {
                        System.out.printf("Sent user: %s, metadata: partition: %s, offset: %s\n", avroRecord, metadata.partition(), metadata.offset());
                    } else {
                        System.out.println("data sent failed: " + exception.getMessage());
                    }
                });
    }

    private Properties getProperties() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Topic.SERVERS_CONFIG);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "BiDemoProducer");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                ByteArraySerializer.class.getName());
        return props;
    }
}
